package com.atguigu.day06;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class Flink02_Window_WindowFun_Agg {
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //2.从端口读取数据并转为WaterSensor
        WindowedStream<WaterSensor, Tuple, TimeWindow> window = env.socketTextStream("localhost", 9999)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new WaterSensor(split[0], Long.parseLong(split[1]), Integer.parseInt(split[2]));
                    }
                })
                //将相同id的数据聚合到一块
                .keyBy("id")
                //开启一个基于处理时间的滚动窗口  窗口大小5S
                .window(ProcessingTimeSessionWindows.withGap(Time.seconds(3)));
        window
                //TODO 使用窗口函数 增量聚合函数Agg 实现对vc求和
                .aggregate(new AggregateFunction<WaterSensor, Integer, String>() {
                    //创建累加器 或者 初始化累加器    在创建窗口的时候为每一个窗口创建一个累加器
                    @Override
                    public Integer createAccumulator() {
                        System.out.println("创建累加器");
                        return 0;
                    }

                    //给累加器赋值 做累加计算  每来一条数据调用一次 相当于来一条数据计算一次
                    @Override
                    public Integer add(WaterSensor value, Integer accumulator) {
                        System.out.println("累加计算");
                        return value.getVc()+accumulator;
                    }

                    //获取结果  等窗口触发计算的时候调用一次 将这个窗口函数计算的结果返回
                    @Override
                    public String getResult(Integer accumulator) {
                        System.out.println("获取结果");
                        return accumulator+"";
                    }

                    //合并累加器 主要用在会话窗口合并的时候 具体是基于事件时间的乱序数据
                    @Override
                    public Integer merge(Integer a, Integer b) {
                        System.out.println("合并累加器");
                        return null;
                    }
                })
                .print();

        window.process(new ProcessWindowFunction<WaterSensor, String, Tuple, TimeWindow>() {
            @Override
            public void process(Tuple tuple, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
                String msg =
                        "窗口: [" + context.window().getStart() / 1000 + "," + context.window().getEnd() / 1000 + ") 一共有 "
                                + elements.spliterator().estimateSize() + "条数据 ";
                out.collect(msg);
            }
        }).print();


        env.execute();
    }
}
